package com.zhu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhu.bean.TradeProvinceOrderWindowBean;
import com.zhu.config.KafkaTopicConfig;
import com.zhu.function.DIMAsyncFunction;
import com.zhu.utils.ClickHouseUtil;
import com.zhu.utils.DataFormatUtil;
import com.zhu.utils.TimestampLtz3CompareUtil;
import com.zhu.utils.ZhuKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * 从 Kafka 读取订单明细数据，过滤 null 数据并按照唯一键对数据去重，统计各省份各
 * 窗口订单数和订单金额，将数据写入 ClickHouse 交易域省份粒度下单各窗口汇总表。
 */
public class DWSTradeProvinceOrderWindowApp {

    public static void main(String[] args) throws Exception {

        //todo env
        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        streamExecutionEnvironment.setParallelism(4);
        //checkPoint
         /*
        streamExecutionEnvironment.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE);
        streamExecutionEnvironment.setStateBackend(new HashMapStateBackend());
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointStorage(ClusterParametersConfig.HDFS_CHECKPOINT_FILE_DIR);
        System.setProperty("HADOOP_USER_NAME","zhu");
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointTimeout(10 * 600000L);
        streamExecutionEnvironment.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        streamExecutionEnvironment.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5 * 1000L));
         */

        //todo kafka_dwd_order_detail
        String topic = KafkaTopicConfig.KAFKA_DWD_TRADE_ORDER_DETAIL_TOPIC;
        String groupId = "dws_trade_province_order_window" + KafkaTopicConfig.KAFKA_GROUP_ID_LAST_NAME;
        DataStreamSource<String> kafkaDStream = streamExecutionEnvironment.addSource(ZhuKafkaUtil.getFlinkKafkaConsumer(topic, groupId));


        //todo
        SingleOutputStreamOperator<JSONObject> jsonObjDStream = kafkaDStream.flatMap(
                new FlatMapFunction<String, JSONObject>() {
                    @Override
                    public void flatMap(String data, Collector<JSONObject> collector) throws Exception {
                        try {
                            JSONObject jsonObject = JSON.parseObject(data);
                            collector.collect(jsonObject);
                        } catch (Exception exception) {
                            System.out.println("dirtyData >>>" + data);
                            exception.printStackTrace();
                        }
                    }
                }
        );
        //todo orderDetail_id 分组 过滤不完整关联数据
        KeyedStream<JSONObject, String> keyedByOrderDetailIDDStream = jsonObjDStream.keyBy(jsonObject -> jsonObject.getString("id"));
        SingleOutputStreamOperator<JSONObject> filterOrderDetailDStream = keyedByOrderDetailIDDStream.process(

                //状态保留整条数据

                new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    private ValueState<JSONObject> lastValueState;


                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<JSONObject> lastJoinStateDescriptor = new ValueStateDescriptor<>("last-join_data", JSONObject.class);
                        lastValueState = getRuntimeContext().getState(lastJoinStateDescriptor);
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, JSONObject, JSONObject>.OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject value = this.lastValueState.value();
                        if (value != null) {
                            out.collect(value);
                        }
                        lastValueState.clear();

                    }

                    @Override
                    public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        JSONObject lastData = lastValueState.value();
                        if (lastData == null) {
                            //注册定时器 写入状态
                            long currentTime = context.timerService().currentProcessingTime();
                            context.timerService().registerProcessingTimeTimer(currentTime + 5000L);
                            lastValueState.update(jsonObject);
                        } else {
                            String lastRowOperatorTime = lastValueState.value().getString("row_op_ts");
                            String thisRowOperatorTime = jsonObject.getString("row_op_ts");
                            if (TimestampLtz3CompareUtil.compare(lastRowOperatorTime, thisRowOperatorTime) <= 0) {
                                lastValueState.update(jsonObject);
                            }
                        }
                    }
                }
        );
        //todo 转换数据结构  为 TradeProvinceOrderWindowBean
        SingleOutputStreamOperator<TradeProvinceOrderWindowBean> beanDStream = filterOrderDetailDStream.map(
                jsonObject -> {
                    String provinceId = jsonObject.getString("province_id");
                    Double splitTotalAmount = jsonObject.getDouble("split_total_amount");
                    String orderId = jsonObject.getString("order_id");
                    Long time = jsonObject.getLong("create_time");

                    return TradeProvinceOrderWindowBean.builder()
                            .provinceId(provinceId)
                            .orderAmount(splitTotalAmount)
                            .orderIdSet(new HashSet<String>(Collections.singleton(orderId)))
                            .ts(time)
                            .build();
                }
        );

        //todo waterMark
        SingleOutputStreamOperator<TradeProvinceOrderWindowBean> provinceWithWaterMarkDStream = beanDStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeProvinceOrderWindowBean>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<TradeProvinceOrderWindowBean>() {
                    @Override
                    public long extractTimestamp(TradeProvinceOrderWindowBean tradeProvinceOrderWindowBean, long l) {
                        return tradeProvinceOrderWindowBean.getTs();
                    }
                }));
        //todo  省份id分组
        KeyedStream<TradeProvinceOrderWindowBean, String> keyedStream = provinceWithWaterMarkDStream.keyBy(TradeProvinceOrderWindowBean::getProvinceId);
        WindowedStream<TradeProvinceOrderWindowBean, String, TimeWindow> windowProvinceDStream = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10)));


        //todo reduce
        SingleOutputStreamOperator<TradeProvinceOrderWindowBean> reduceDetailByProvinceDStream = windowProvinceDStream.reduce(
                new ReduceFunction<TradeProvinceOrderWindowBean>() {
                    @Override
                    public TradeProvinceOrderWindowBean reduce(TradeProvinceOrderWindowBean value1, TradeProvinceOrderWindowBean value2) throws Exception {
                        value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                        value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());
                        return value1;
                    }
                },
                new ProcessWindowFunction<TradeProvinceOrderWindowBean, TradeProvinceOrderWindowBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<TradeProvinceOrderWindowBean, TradeProvinceOrderWindowBean, String, TimeWindow>.Context context, Iterable<TradeProvinceOrderWindowBean> iterable, Collector<TradeProvinceOrderWindowBean> collector) throws Exception {
                        String start = DataFormatUtil.toYmdHms(context.window().getStart());
                        String end = DataFormatUtil.toYmdHms(context.window().getEnd());

                        TradeProvinceOrderWindowBean bean = iterable.iterator().next();
                        bean.setStt(start);
                        bean.setEdt(end);

                        bean.setTs(System.currentTimeMillis());
                        bean.setOrderCount((long) bean.getOrderIdSet().size());
                        collector.collect(bean);

                    }
                }

        );

        //todo 关联省份信息 province_id
        SingleOutputStreamOperator<TradeProvinceOrderWindowBean> fullInfoStream = AsyncDataStream.unorderedWait(reduceDetailByProvinceDStream,
                new DIMAsyncFunction<TradeProvinceOrderWindowBean>("DIM_BASE_PROVINCE") {
                    @Override
                    public void join(TradeProvinceOrderWindowBean input, JSONObject dimInfo) {
                        input.setProvinceName(dimInfo.getString("NAME"));
                    }

                    @Override
                    public String getKey(TradeProvinceOrderWindowBean input) {
                        return input.getProvinceId();
                    }
                }, 100, TimeUnit.SECONDS);

        //todo 写入到OLAP
        fullInfoStream.print("我一定要去深圳!!!!");


        fullInfoStream.addSink(ClickHouseUtil.getClickHouseSinkFunction(
                "insert into dws_trade_province_order_window values(?,?,?,?,?,?,?)"
        ));


        //todo job
        streamExecutionEnvironment.execute();


    }

}
